package com.gxlevi.task

import com.gxlevi.bean.ClickLogWide
import org.apache.commons.lang3.StringUtils
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow


trait BaseTask[T] {

  def map(clickWideDataStream: DataStream[ClickLogWide]): DataStream[T]

  def keyBy(mapStream: DataStream[T]): KeyedStream[T, String]

  def timeWindow(keyedStream: KeyedStream[T, String]): WindowedStream[T, String, TimeWindow] = {
    keyedStream.timeWindow(Time.seconds(3))
  }

  def reduce(windowedStream: WindowedStream[T, String, TimeWindow]): DataStream[T]

  def sinkToHBase(reduceDataStream: DataStream[T])

  def process(clickWideDataStream: DataStream[ClickLogWide]) = {
    val mapDataStream: DataStream[T] = map(clickWideDataStream)
    val keyedStream = keyBy(mapDataStream)
    val windowedStream = timeWindow(keyedStream)
    val reduceStream = reduce(windowedStream)
    sinkToHBase(reduceStream)
  }

  val isOld = (isNew: Int, isDateNew: Int) => if (isNew == 0 && isDateNew == 1) 1 else 0

  var tableName = ""
  var clfName = "info"
  var rowKey = ""
  val channelIdColumn = "channelId"
  val dateColumn = "date"
  val pvColumn = "pv"
  val uvColumn = "uv"
  val newCountColumn = "newCount"
  val oldCountColumn = "oldCount"

  def getTotal(resultMap: Map[String, String], column: String, currentValue: Long): Long = {
    var total = 0L
    if (resultMap != null && StringUtils.isNotBlank(resultMap.getOrElse(column, ""))) {
      total = resultMap(column).toLong + currentValue
    } else {
      total = currentValue
    }
    total
  }
}
